import com.zhenjun.domin.Item;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSink;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.ArrayList;
import java.util.List;

/**
 * @author wangzj
 * @description 聚合函数的使用
 * @date 2020/7/12 0:31
 */
public class AggregationsDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //获取数据源
        List data = new ArrayList<Tuple3<Integer, Integer, Integer>>();
        data.add(new Tuple3<>(0, 1, 0));
        data.add(new Tuple3<>(0, 1, 1));
        data.add(new Tuple3<>(0, 1, 3));
        data.add(new Tuple3<>(0, 2, 2));
        data.add(new Tuple3<>(1, 2, 5));
        data.add(new Tuple3<>(1, 2, 11));
        data.add(new Tuple3<>(1, 1, 9));
        data.add(new Tuple3<>(1, 2, 13));

        //将list数据转化成DataStream
        DataStreamSource<Item> text = env.fromCollection(data);
        //第一个元素进行聚合，并且按照第三个元素取最大值
        /**
         * 3> (0,1,0)
         * 3> (0,1,1)
         * 3> (0,1,2)
         * 3> (0,1,3)
         * 3> (1,2,5)
         * 3> (1,2,9)
         * 3> (1,2,11)
         * 3> (1,2,13)
         */
//        text.keyBy(0).max(2).printToErr();

        /**
         * 3> (0,1,0)
         * 3> (0,1,1)
         * 3> (0,2,2)
         * 3> (0,1,3)
         * 3> (1,2,5)
         * 3> (1,2,9)
         * 3> (1,2,11)
         * 3> (1,2,13)
         */
//        text.keyBy(0).maxBy(2).printToErr();

        //聚合操作
        DataStreamSource<Tuple3<Integer,Integer,Integer>> text1 = env.fromCollection(data);
        DataStreamSink<Tuple3<Integer, Integer, Integer>> reduce = text1
                .keyBy(0)
                .reduce(new ReduceFunction<Tuple3<Integer, Integer, Integer>>() {
                    @Override
                    public Tuple3<Integer, Integer, Integer> reduce(Tuple3<Integer, Integer, Integer> t1, Tuple3<Integer, Integer, Integer> t2) throws Exception {
                        Tuple3<Integer, Integer, Integer> newTuple = new Tuple3<>();

                        newTuple.setFields(0, 0, (Integer) t1.getField(2) + (Integer) t2.getField(2));
                        return newTuple;
                    }
                })
                .printToErr();


        env.execute("AggregationsDemo");
    }
}
